package com.pw.study.flink.window

import com.pw.study.flink.source.MyWordSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**
 * @Desc: TODO
 * @author: future
 * @since: 2022/5/15-15:42
 */
object MyTimeWindow {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val ds: DataStream[String] = env.addSource(new MyWordSource())
    val dsKey: KeyedStream[(String, Int), String] = ds.map((_, 1)).keyBy(_._1)
    val dsWindow: WindowedStream[(String, Int), String, TimeWindow] = dsKey.timeWindow(Time.seconds(3))
    val dsSum: DataStream[(String, Int)] = dsWindow.sum(1)

    val dsSlide: DataStream[(String, Int)] = dsKey.timeWindow(Time.seconds(3), Time.seconds(1)).sum(1)

    dsSlide.print("Slide>>")
    dsSum.print("window>>")
    env.execute("timeWindow")
  }
}
